package com.lwd.util.collection;


import com.lwd.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.function.Function;

/**
 * @author Brozen
 * @date 2019/7/11 10:09 AM
 */
public class SubCollectionExecutor {

    /**
     * 按指定间隔将时间段分隔执行
     *  @param src 源集合
     * @param batchSize 批量处理条数
     */
    public static <E> List<Future<?>> run(ExecutorService executor, final Consumer<Collection<E>> runnable, Collection<E> src, int batchSize) {
        List<Future<?>> futures = new LinkedList<>();
        List<E> subCollection = new ArrayList<>(batchSize);

        for (E ele : src) {
            subCollection.add(ele);
            if (subCollection.size() >= batchSize) {
                final List<E> finalSubCollection = subCollection;
                subCollection = new ArrayList<>(batchSize);
                futures.add(executor.submit(new Runnable() {
                    @Override
                    public void run() {
                        runnable.accept(finalSubCollection);
                    }
                }));
            }
        }

        if (CollectionUtils.isNotEmpty(subCollection)) {
            final List<E> finalSubCollection = subCollection;
            futures.add(executor.submit(new Runnable() {
                @Override
                public void run() {
                    runnable.accept(finalSubCollection);
                }
            }));
        }

        return futures;
    }

    public static <IN, OUT> List<Future<OUT>> call(ExecutorService executor, final Function<Collection<IN>, OUT> callback,
                                                   Collection<IN> src, int batchSize) {
        List<Future<OUT>> futures = new LinkedList<>();
        List<IN> subCollection = new ArrayList<>(batchSize);

        for (IN ele : src) {
            subCollection.add(ele);
            if (subCollection.size() >= batchSize) {
                final List<IN> finalSubCollection = subCollection;
                subCollection = new ArrayList<>(batchSize);
                futures.add(executor.submit(new Callable<OUT>() {
                    @Override
                    public OUT call() {
                        return callback.apply(finalSubCollection);
                    }
                }));
            }
        }

        if (CollectionUtils.isNotEmpty(subCollection)) {
            final List<IN> finalSubCollection = subCollection;
            futures.add(executor.submit(new Callable<OUT>() {
                @Override
                public OUT call() {
                    return callback.apply(finalSubCollection);
                }
            }));
        }

        return futures;
    }

    public static <E> void run(Consumer<Collection<E>> runnable, Collection<E> src, int batchSize) {
        List<E> subCollection = new ArrayList<>(batchSize);

        for (E ele : src) {
            subCollection.add(ele);
            if (subCollection.size() >= batchSize) {
                runnable.accept(subCollection);
                subCollection = new ArrayList<>(batchSize);
            }
        }

        if (CollectionUtils.isNotEmpty(subCollection)) {
            runnable.accept(subCollection);
        }
    }

    // 等待future执行完成；若其中一个future报错，则尝试停止其余future
    public static void wait(List<Future<?>> futures) {
        Iterator<Future<?>> iterator = futures.iterator();
        try {
            while (iterator.hasNext()) {
                iterator.next().get();
            }
        } catch (InterruptedException | ExecutionException e) {
            while (iterator.hasNext()) {
                iterator.next().cancel(true);
            }
            // 并发报错
            throw new RuntimeException(e instanceof ExecutionException ? e.getCause() : e);
        }
    }
}
